package com.archstudy.springcloud.message.service.impl.messageDispatch;

import com.archstudy.push.platform.contract.dto.MsgDTO;
import com.archstudy.springcloud.common.util.JsonUtil;
import com.archstudy.springcloud.message.service.impl.MessagePushServiceImpl;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;
import org.apache.kafka.clients.consumer.ConsumerRecord;

class MsgEventHandler implements EventHandler<MsgEvent>, WorkHandler<MsgEvent> {

    private MessagePushServiceImpl messagePushService;

    public MsgEventHandler(MessagePushServiceImpl messagePushService) {
        this.messagePushService = messagePushService;
    }

    @Override
    public void onEvent(MsgEvent event) throws Exception {
        ConsumerRecord record = event.getConsumerRecord();
        String value = (String) record.value();
        MsgDTO dto = JsonUtil.jsonToPojo(value, MsgDTO.class);
        messagePushService.saveAndDispatchMessage(dto);
    }

    @Override
    public void onEvent(MsgEvent event, long sequence, boolean endOfBatch) throws Exception {

    }
}
